package com.qushiyun.devops.admin.sync.job.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.qushiyun.devops.admin.sync.data.mapper.TableLogMapper;
import com.qushiyun.devops.admin.sync.data.po.TableLog;
import com.qushiyun.devops.admin.sync.data.po.TriggerJob;
import com.qushiyun.devops.admin.sync.job.SyncService;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map.Entry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.anyline.data.jdbc.ds.DataSourceHolder;
import org.anyline.data.param.ConfigStore;
import org.anyline.data.param.init.DefaultConfigStore;
import org.anyline.entity.Compare;
import org.anyline.entity.DataRow;
import org.anyline.entity.DataSet;
import org.anyline.entity.Order.TYPE;
import org.anyline.metadata.Column;
import org.anyline.metadata.Table;
import org.anyline.proxy.CacheProxy;
import org.anyline.service.AnylineService;
import org.anyline.util.BasicUtil;
import org.springframework.stereotype.Component;

/**
 * 一次性更新同步，主要根据表的update_time(date_time)字段判断是否有新增
 *
 * @author zhouzx
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class UpdateSyncServiceImpl implements SyncService {

    private final AnylineService service;
    private final TableLogMapper tableLogMapper;

    public static final String UPDATE_COLUMN = "update_time";
    public static final String INSERT_COLUMN = "create_time";

    @Override
    public TriggerJob sync(TriggerJob job, String fromDb, String toDb, DataRow conditionRow, String tenantId) throws Exception {
        job.setTriggerTime(new Date());
        DataSourceHolder.setDefaultDataSource();
        List<String> tables = (List<String>) conditionRow.getList("table_list");
        Collection<String> ignores = (List) conditionRow.getList("ignore_table");
        Collection<String> noDataTable = (List) conditionRow.getList("no_data_table");

        if (CollUtil.isEmpty(tables)) {
            log.error("未设置源表:{}", conditionRow.toJSON());
            return job;
        }
        DataSourceHolder.setDataSource(fromDb);
        if (tables.size() == 1 && tables.get(0).equals("*")) {
            CacheProxy.clear();
            tables = service.tables();
        } else {
            CacheProxy.clear();
            List<String> dbTables = service.tables();
            List<String> newTables = new ArrayList<>();
            for (String dbTable : dbTables) {
                for (String table : tables) {
                    if (dbTable.contains(table)) {
                        newTables.add(dbTable);
                    }
                }
            }
            tables = newTables;
        }

        long index = 0, dataNum = 0;
        int tableNum = 0;
        for (String table : tables) {
            long start = System.currentTimeMillis();
            index++;
            log.info("开始同步第 {} 张表 {}", index, table);
            TableLog tableLog = new TableLog();
            tableLog.setBeginTime(new Date());
            boolean ignored = false;
            if (CollUtil.isNotEmpty(ignores)) {
                for (String ignore : ignores) {
                    if (table.contains(ignore)) {
                        ignored = true;
                        break;
                    }
                }
                if (ignored) {
                    tableLog.setIsIgnore(1);
                    tableLog.setEndTime(new Date());
                    tableLog.setCostTime((System.currentTimeMillis() - start));
                    tableLogMapper.insert(tableLog);
                    continue;
                }
            }
            tableNum++;

            if (CollUtil.isNotEmpty(noDataTable)) {
                for (String ignore : noDataTable) {
                    if (table.contains(ignore)) {
                        ignored = true;
                        break;
                    }
                }
                if (ignored) {
                    log.warn("{}只同步表结构，不同步数据", table);
                    tableLog.setDataNum(0L);
                    tableLog.setStatus(2);
                    tableLog.setIsCreated(1);
                    tableLog.setEndTime(new Date());
                    tableLog.setCostTime(System.currentTimeMillis() - start);
                    tableLogMapper.insert(tableLog);
                    continue;
                }
            }

            DataSourceHolder.setDataSource(fromDb);
            conditionRow.put("trigger_id", job.getId());
            conditionRow.put("SRC_TABLE", table);
            conditionRow.put("TAR_TABLE", table);
            conditionRow.put("tenant_id", tenantId);
            conditionRow.put("COLS", "*");

            long size = update(fromDb, toDb, conditionRow);

            dataNum += size;
            log.info("同步表:{}，同步数据量：{}，耗时：{}", table, size, System.currentTimeMillis() - start);
            DataSourceHolder.setDataSource(fromDb);
        }
        job.setDataNum(dataNum);
        job.setTableNum(tableNum);
        return job;
    }

    /**
     * 全量同步
     *
     * @param task 同步信息
     */
    private long update(String fromDb, String toDb, DataRow task) throws Exception {
        List<String> all = (List) task.getList("all_table");
        List<String> tenantAndGlobal = (List) task.getList("tenant_global_table");
        long triggerId = task.getLong("trigger_id");
        String conditionStr = task.getString("condition_str");
        int conditionType = task.getInt("type");
        String tenantId = task.getString("tenant_id");

        if (conditionType == 1 && StrUtil.isBlank(tenantId)) {
            log.error("租户ID为空，直接返回");
            return 0;
        }
        List<String> tenants = new ArrayList<>();
        if (tenantId.contains(",")) {
            tenants = StrUtil.split(tenantId, ",");
        } else {
            tenants.add(tenantId);
        }
        int step = task.getInt("STEP", 2000);
        String srcTable = task.getString("SRC_TABLE");

        Date lastTime = tableLogMapper.getLastSyncTime(srcTable, triggerId);

        String orderKey = task.getStringNvl("ORDER_COLUMN");
        TableLog tableLog = new TableLog();
        tableLog.setBeginTime(new Date());
        tableLog.setTriggerId(task.getLong("trigger_id"));
        long start = System.currentTimeMillis();
        tableLog.setTableName(srcTable);
        tableLog.setIsCreated(1);
        if (BasicUtil.isEmpty(orderKey)) {
            DataSourceHolder.setDataSource(fromDb);
            Table table = service.metadata().table(srcTable);
            LinkedHashMap<String, Column> columns = table.getColumns();
            boolean hasTenantIdColumn = false;
            boolean hasCreateTimeColumn = false;
            for (Entry<String, Column> entry : columns.entrySet()) {
                String k = entry.getKey();
                if (k.equalsIgnoreCase(UPDATE_COLUMN)) {
                    hasCreateTimeColumn = true;
                }
                if (k.equalsIgnoreCase("tenant_id")) {
                    hasTenantIdColumn = true;
                }
            }

            if (!hasCreateTimeColumn) {
                log.warn("{} 表没有update_time字段，无法一次性更新同步！", srcTable);
                return 0;
            }

            if (!hasTenantIdColumn && conditionType == 1) {
                log.error("{} 表没有租户Id", srcTable);
                //DataSourceHolder.setDataSource(fromDb);
                //service.execute(String.format(ADD_TENANT_ID, srcTable, orderKey));
            }
            Column col = table.primary();
            if (null != col) {
                orderKey = col.getName();
            }
            String max = "0";
            boolean allSync = false;
            if (CollUtil.isNotEmpty(all)) {
                for (String allTableName : all) {
                    if (srcTable.contains(allTableName)) {
                        allSync = true;
                        break;
                    }
                }
            }

            if (allSync) {
                tableLog.setIsAll(1);
                DataSourceHolder.setDataSource(toDb);
                service.execute("TRUNCATE TABLE " + srcTable);
                DataSourceHolder.setDataSource(fromDb);
                DataSet set = service.querys(srcTable);
                if (set.size() != 0) {
                    //set = format(set, cols);
                    DataSourceHolder.setDataSource(toDb);
                    service.insert(srcTable, set);
                }

                tableLog.setDataNum((long) set.size());
                tableLog.setStatus(2);
                tableLog.setEndTime(new Date());
                tableLog.setCostTime(System.currentTimeMillis() - start);
                tableLogMapper.insert(tableLog);
                return set.size();
            }

            ConfigStore condition = new DefaultConfigStore(0, step - 1);
            //ConfigStore updateStore = new DefaultConfigStore();
            //updateStore.gt(UPDATE_COLUMN, lastTime);
            condition.gt(UPDATE_COLUMN, lastTime);
            if (conditionType == 1) {
                if (CollUtil.isNotEmpty(tenantAndGlobal) && tenantAndGlobal.contains(srcTable)) {
                    tenants.add("-1");
                    condition.and("tenant_id", tenants);
                    tableLog.setIsTenantGlobal(1);
                } else {
                    condition.and("tenant_id", tenants);
                }
            } else if (conditionType == 2 && StrUtil.isNotBlank(conditionStr)) {
                condition.and(conditionStr);
            }
            long exeQty = 0;
            while (true) {
                DataSourceHolder.setDataSource(fromDb);
                condition.and(Compare.GREAT, orderKey, max);
                condition.order(orderKey, TYPE.ASC);
                DataSet set = service.querys(srcTable, condition);
                max = set.getString(set.size() - 1, orderKey, "0");
                if (set.size() == 0) {
                    break;
                }
                exeQty += set.size();
                //set = format(set, cols);
                DataSourceHolder.setDataSource(toDb);
                Object last = set.get(set.size() - 1, INSERT_COLUMN);
                if (last instanceof Timestamp) {
                    Timestamp timestamp = (Timestamp) last;
                    LocalDateTime localDateTime = timestamp.toLocalDateTime();
                    tableLog.setLastSyncTime(localDateTime);
                } else if (last instanceof LocalDateTime) {
                    tableLog.setLastSyncTime((LocalDateTime) last);
                } else {
                    log.warn("unknown update_time type {}", last);
                }
                set.setOverride(true);
                set.setPrimaryKey("id");
                set.addAllUpdateColumns();
                service.update(srcTable, set);
            }
            tableLog.setDataNum(exeQty);
            tableLog.setStatus(2);
            tableLog.setEndTime(new Date());
            tableLog.setCostTime(System.currentTimeMillis() - start);
            tableLogMapper.insert(tableLog);
            return exeQty;
        }
        return 0;
    }
}
